package com.lxl.testHd.etl.two;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.CombineTextInputFormat;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import java.io.IOException;
import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.Date;

/**
 * @author ：e_lixilin
 * @date ：2022/2/22 9:53
 * @description：
 * @modified By：
 */
public class MyTollStation {

    static class CarDriveMapper extends Mapper<LongWritable, Text, CarInfo, NullWritable> {
        CarInfo carInfo = new CarInfo();
        NullWritable outValue = NullWritable.get();
        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            // 2018-09-15.06:31:31	2018-09-15.08:58:58	 262	中型	浙KL69T0
            String line = value.toString();
            String[] words = line.split("\t");
            // 车牌
            carInfo.setCarPlate(words[words.length - 1]);
            // 行驶里程
            long distance = Long.parseLong(words[2]);
            carInfo.setDistance(distance);
            SimpleDateFormat sdf = new SimpleDateFormat("yyyy-MM-dd.HH:mm:ss") ;
            //进站时间
            String startDateStr = words[0] ;
            //出站时间
            String endDateStr = words[1] ;
            Date startDate = null ;
            Date endDate = null ;
            try {
                startDate = sdf.parse(startDateStr);
                endDate = sdf.parse(endDateStr);
            } catch (ParseException e) {
                System.out.println(" 时间解析错误 ");
            }
            // 行驶时间
            long minutes = (endDate.getTime() - startDate.getTime()) / 1000 / 60;
            carInfo.setMinutes(minutes);
            // 行驶速度
            double speed = distance *1.0/(minutes/60);
            carInfo.setSpeed(speed);
            context.write(carInfo,outValue);
        }
    }

    static class CarDriveReduce extends Reducer<CarInfo, NullWritable,Text, NullWritable> {
        Text outKey = new Text();
        StringBuilder sb = new StringBuilder();
        NullWritable outVal = NullWritable.get();
        @Override
        protected void reduce(CarInfo key, Iterable<NullWritable> values, Context context) throws IOException, InterruptedException {
            sb.setLength(0);
            sb.append(key.getCarPlate());
            sb.append("\t");
            sb.append(key.getDistance());
            sb.append("\t");
            sb.append(key.getMinutes());
            sb.append("\t");
            sb.append(key.getSpeed());

            outKey.set(sb.toString());
            context.write(outKey,outVal);
        }
    }
    public static void main(String[] args) throws Exception {
        args = new String[]{"D:\\big-data\\etl", "D:\\big-data\\etl\\outputetltwo"};
        Configuration conf = new Configuration();
        Job job = Job.getInstance(conf);
        job.setJarByClass(MyTollStation.class);
        job.setMapperClass(CarDriveMapper.class);
        job.setReducerClass(CarDriveReduce.class);
        //修改输出类型
        job.setMapOutputKeyClass(CarInfo.class);
        job.setMapOutputValueClass(NullWritable.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(CarInfo.class);
        job.setGroupingComparatorClass(ProvinceGroup.class);

        job.setInputFormatClass(CombineTextInputFormat.class);
        /*
         * 设置切片的大小介于 100M-128M之间。注意，这里是指：当多个小文件合并后，如果正好介于 100M-128M，
         * 就将这些小文件合并。但是，如果全部的小文件合并后仍然<100MB，也是会将全部小文件合并成一个切片。
         */
        CombineTextInputFormat.setMinInputSplitSize(job,100*1024*1024 );
        CombineTextInputFormat.setMaxInputSplitSize(job,128*1024*1024 );

        FileInputFormat.setInputPaths(job, new Path(args[0]));
        FileOutputFormat.setOutputPath(job, new Path(args[1]));
        boolean res = job.waitForCompletion(true);
        System.exit(res?0:1);
    }
}
